/*
 * Copyright 2011-2012 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.sabayframework.mem.impl;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.sabayframework.channel.Channel;
import org.sabayframework.channel.ram.ArrayChannel;
import org.sabayframework.mem.BigBuffer;
import org.sabayframework.mem.IdGenerator;
import org.sabayframework.mem.Space;
import org.sabayframework.mem.Transaction;
import org.sabayframework.mem.Transaction.Status;
import org.sabayframework.mem.TransactionNotFoundException;
import org.sabayframework.queue.ExecutionQueue;

public abstract class AbstractSpace implements Space {

	private final IdGenerator idGenerator;
	private final ExecutionQueue executionQueue;
	private final HashMap<Long, Transaction> activeTransactions;
	private final ReentrantReadWriteLock lock;
	
	private volatile long lastCommitted;
	private volatile boolean closed;
	
	public AbstractSpace(ExecutionQueue executionQueue) {
		this(new SequentialIdGenerator(0), executionQueue);
	}
			
	/**
	 * @param idGenerator id generator for transactions 
	 * @param executionQueue 	
	 */
	public AbstractSpace(
			IdGenerator idGenerator, 
			ExecutionQueue executionQueue) {
		super();
		this.idGenerator = idGenerator;
		this.executionQueue = executionQueue;
		this.activeTransactions = new HashMap<Long, Transaction>();
		this.lock = new ReentrantReadWriteLock();
	}

	@Override
	public Channel<Transaction> getTransaction(long txId) {
		ArrayChannel<Transaction> res = new ArrayChannel<Transaction>(1);
		Transaction t = null;
		lock.readLock().lock();
		try{
			t = activeTransactions.get(txId);
		} finally{
			lock.readLock().unlock();
		}
		if(t==null) res.fail(new TransactionNotFoundException(String.valueOf(txId)));
		else {
			res.add(t);
			res.close();
		}
		return res;
	}
	
	@Override
	public IdGenerator getIdGenerator() {
		return idGenerator;
	}
	
	public void started(TransactionImpl tx) {
		lock.writeLock().lock();
		try{
			activeTransactions.put(tx.getId(), tx);
		} finally{
			lock.writeLock().unlock();
		}
	}

	public void completed(TransactionImpl tx) throws Exception {
		final long txid = tx.getId();
		final boolean committed = tx.getStatus() == Status.COMMIT;
		lock.writeLock().lock();
		try{
			activeTransactions.remove(txid);
			if(committed) lastCommitted = txid;
		} finally{
			lock.writeLock().unlock();
		}
	}

	@Override
	public void close() throws IOException {
		closed = true;
	}

	public boolean isClosed() {
		return closed;
	}

	@Override
	public ExecutionQueue getExecutionQueue() {
		return executionQueue;
	}

	/**
	 * Returns id of last committed transaction.
	 * */
	public long getLastCommittedId() {
		return lastCommitted;
	}

	/***
	 * Returns read-write lock for this space. 
	 */
	public ReentrantReadWriteLock getLock() {
		return lock;
	}
	
	@Override
	public BigBuffer wrap(final byte[] array,final int offset,final int length) {
		if(array == null || length == 0) return EmptyBigBuffer.INSTANCE;
		return new UnsafeByteArrayBuffer(array, offset, length);
	}

	@Override
	public BigBuffer wrap(ByteBuffer buffer) {
		if(buffer == null || buffer.capacity() == 0) return EmptyBigBuffer.INSTANCE;
		return new BigByteBuffer(buffer);
	}

	@Override
	public BigBuffer allocateDirect(final long capacity) {
		if(capacity == 0) return EmptyBigBuffer.INSTANCE;
		return new UnsafeDirectBuffer(capacity);
	}

	@Override
	public BigBuffer allocate(final int capacity) {
		if(capacity == 0) return EmptyBigBuffer.INSTANCE;
		return new UnsafeByteArrayBuffer(new byte[capacity], 0, capacity);
	}

	@Override
	protected void finalize() throws Throwable {
		super.finalize();
		close();
	}

}
